package com.onlinelearn.notifyservice.event;

import com.onlinelearn.notifyservice.model.Notification;
import com.onlinelearn.notifyservice.model.NotificationType;
import com.onlinelearn.notifyservice.service.NotificationService;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;

import java.io.Serializable;

public class NotificationEvent extends ApplicationEvent implements Serializable {
    private final Notification notification;
    private final Long userId;
    private final NotificationType type;
    private final String content;

    public NotificationEvent(Object source, Notification notification) {
        super(source);
        this.notification = notification;
        this.userId = notification.getUserId();
        this.type = notification.getType();
        this.content = notification.getContent();
    }

    public Notification getNotification() {
        return notification;
    }

    public Long getUserId() {
        return userId;
    }

    public NotificationType getType() {
        return type;
    }

    public String getContent() {
        return content;
    }
}

@Component
public class NotificationEventPublisher {
    private final ApplicationEventPublisher eventPublisher;
    private final KafkaTemplate<String, NotificationEvent> kafkaTemplate;

    @Autowired
    public NotificationEventPublisher(
        ApplicationEventPublisher eventPublisher,
        KafkaTemplate<String, NotificationEvent> kafkaTemplate
    ) {
        this.eventPublisher = eventPublisher;
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publishNotificationEvent(Notification notification) {
        NotificationEvent event = new NotificationEvent(this, notification);
        eventPublisher.publishEvent(event);
        kafkaTemplate.send("notification-topic", event);
    }
}

@Component
public class NotificationEventConsumer {
    private final NotificationService notificationService;

    @Autowired
    public NotificationEventConsumer(NotificationService notificationService) {
        this.notificationService = notificationService;
    }

    @KafkaListener(topics = "notification-topic")
    public void handleNotificationEvent(NotificationEvent event) {
        notificationService.sendNotification(
            event.getUserId(), 
            event.getType(), 
            event.getContent()
        );
    }

    @EventListener
    public void handleLocalNotificationEvent(NotificationEvent event) {
        notificationService.processLocalNotification(event.getNotification());
    }
} 